package qq2564874169.tuaotuao.fx.mq.zookeeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import qq2564874169.tuaotuao.fx.FxAssert;
import qq2564874169.tuaotuao.fx.FxStream;
import qq2564874169.tuaotuao.fx.mq.MqPublisher;
import qq2564874169.tuaotuao.fx.utils.ExUtil;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class ZkMqPublisher implements MqPublisher {
    private static String rootNode;
    private ZooKeeper zk;

    public static void createRootNode(ZooKeeper zk) {
        createRootNode(zk, "/FxMsgQueue");
    }

    public static void createRootNode(ZooKeeper zk, String nodeName) {
        try {
            if (zk.exists(nodeName, false) == null) {
                zk.create(nodeName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            rootNode = nodeName;
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    private static String fullPath(String path) {
        FxAssert.isNotNull(path);
        FxAssert.isNotNull(rootNode, "请先调用createRootNode创建根节点");
        if (path.startsWith("/")) {
            return rootNode + path;
        } else {
            return rootNode + "/" + path;
        }
    }

    public ZkMqPublisher(ZooKeeper zk) {
        this.zk = zk;
    }

    @Override
    public void close() {

    }

    @Override
    public void post(String queue, List<String> msg) {
        if (msg.isEmpty()) {
            return;
        }
        try {
            var msglist = FxStream.create(msg).map(i -> i.getBytes(StandardCharsets.UTF_8)).toList();
            var fk = fullPath(queue);
            var childs = zk.getChildren(fk, false);
            for (String c : childs) {
                var nkey = fk + "/" + c;
                var msgkey = nkey + "/" + ZkMqSubscriber.msgNode;
                for (var buf : msglist) {
                    zk.create(msgkey, buf, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
                }
            }
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }
}
